package edu.hhu.innerac.sparkm.demo;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;
import scala.reflect.internal.Trees.If;
import edu.hhu.innerac.sparkm.SparkProgress;
import edu.hhu.innerac.sparkm.entry.BasePhone;

public class InStoreRate implements Serializable{

	public List<BasePhone> genData(){
		List<BasePhone> phones = new ArrayList<BasePhone>();
		
		phones.add(new BasePhone("A1:B1:C1", 1000000, true, false));
		phones.add(new BasePhone("A1:B1:C1", 1000001, true, false));
		phones.add(new BasePhone("A1:B1:C1", 1000002, true, false));
		phones.add(new BasePhone("A1:B1:C1", 1000003, true, false));
		phones.add(new BasePhone("A1:B1:C1", 1000004, true, false));
		phones.add(new BasePhone("A1:B1:C1", 1000005, false, true));
		phones.add(new BasePhone("A1:B1:C1", 1000006, false, true));
		phones.add(new BasePhone("A1:B1:C1", 1000007, false, true));
		
		phones.add(new BasePhone("A2:B2:C2", 1000000, false, true));
		phones.add(new BasePhone("A2:B2:C2", 1000001, false, true));
		phones.add(new BasePhone("A2:B2:C2", 1000002, false, true));
		phones.add(new BasePhone("A2:B2:C2", 1000003, false, true));
		phones.add(new BasePhone("A2:B2:C2", 1000004, false, true));
		phones.add(new BasePhone("A2:B2:C2", 1000005, false, true));
		phones.add(new BasePhone("A2:B2:C2", 1000006, false, true));
		phones.add(new BasePhone("A2:B2:C2", 1000007, false, true));
		
		phones.add(new BasePhone("A3:B3:C3", 1000000, false, true));
		phones.add(new BasePhone("A3:B3:C3", 1000001, false, true));
		phones.add(new BasePhone("A3:B3:C3", 1000002, false, true));
		phones.add(new BasePhone("A3:B3:C3", 1000003, false, true));
		phones.add(new BasePhone("A3:B3:C3", 1000004, false, true));
		phones.add(new BasePhone("A3:B3:C3", 1000005, true, true));
		phones.add(new BasePhone("A3:B3:C3", 1000006, true, true));
		phones.add(new BasePhone("A3:B3:C3", 1000007, true, true));
		
		
		
		return phones;
	}
	
	public void start(){
		SparkConf sparkConf = new SparkConf().setAppName("instoreRate").setMaster("local");
		JavaSparkContext jsc = new JavaSparkContext(sparkConf);
		List<BasePhone> phones = genData();
		
		JavaRDD<BasePhone> phoneRDD = jsc.parallelize(phones);
		System.out.println(phoneRDD.count());
		phoneRDD.map(new Function<BasePhone, String>() {

			public String call(BasePhone arg0) throws Exception {
				return null;
			}
		});
		
		JavaPairRDD<String, Integer> phonep = phoneRDD.mapToPair(new PairFunction<BasePhone, String, Integer>() {

			public Tuple2<String, Integer> call(BasePhone arg0)
					throws Exception {
				long time = arg0.getTime() - arg0.getTime()%3;
				if(arg0.isInner()){
					return new Tuple2<String, Integer>(arg0.getMac()+","+time, 2);
				}else if(arg0.isOuter()){
					return new Tuple2<String, Integer>(arg0.getMac()+","+time, 1);
				}else{
					return new Tuple2<String, Integer>(arg0.getMac()+","+time, 0);
				}
			}
		});
		
		JavaPairRDD<String, Integer> res0 = phonep.reduceByKey(new Function2<Integer, Integer, Integer>() {

			public Integer call(Integer arg0, Integer arg1) throws Exception {
				return arg0>arg1?arg0:arg1;
			}
		});
		
		JavaPairRDD<String, Integer> res1 = res0.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Integer>() {

			public Tuple2<String, Integer> call(Tuple2<String, Integer> t)
					throws Exception {
				
				return new Tuple2<String, Integer>(t._1.split(",")[1], t._2);
			}
		});
		
//		JavaPairRDD<String, Integer> res2 = res1.reduceByKey(new Function2<Integer, Integer, Integer>() {
//
//			public Integer call(Integer arg0, Integer arg1) throws Exception {
//				return arg0>arg1?arg0:arg1;
//			}
//		});
		
//		System.out.println(res1.collect());
//		System.out.println(res1.countByValue());
		Map<Tuple2<String, Integer>, Long> res2 = res1.countByValue();
		for(Tuple2<String, Integer> keys : res2.keySet()){
			System.out.println(keys._1 + (keys._2 == 1?" 门外 ":" 门内 ")+res2.get(keys));
		}
		jsc.stop();
	}
	
	public static void main(String args[]){
		InStoreRate inStoreRate = new InStoreRate();
		inStoreRate.start();
	}

}
